package broker

import (
	"errors"
	"gitee.com/tjccc/gomq/lib/logger"
	conn "gitee.com/tjccc/gomq/mq/connection"
	"sync"
	"sync/atomic"
)

type GetMsgReq struct {
	conn conn.Connection
}

type GetMsgResp struct {
	data  [][]byte
	index int
}

type ConsumerGroup struct {
	Id         int
	GroupName  string
	Topic      map[string]atomic.Bool
	Consumer   map[int32]conn.Connection
	msgReqChan chan *GetMsgReq
	sync.Mutex
}

func NewConsumerGroup(name string) *ConsumerGroup {
	return &ConsumerGroup{
		GroupName:  name,
		Consumer:   map[int32]conn.Connection{},
		msgReqChan: make(chan *GetMsgReq, 1024),
	}
}

func (cg *ConsumerGroup) AddConsumer(client conn.Connection) {
	cg.Lock()
	defer cg.Unlock()
	cg.Consumer[client.GetClientId()] = client
}

func (cg *ConsumerGroup) NotifyMsg(topicName string) {
	cg.Lock()
	defer cg.Unlock()
	sub, ok := cg.Topic[topicName]
	if !ok {
		logger.Error("topic[%s] has no subscription", topicName)
		return
	}
	sub.Store(true)
}

func (cg *ConsumerGroup) PullMsg(clientId int32) error {
	client, ok := cg.Consumer[clientId]
	if !ok {
		return errors.New("client not exist")
	}
	cg.msgReqChan <- &GetMsgReq{
		conn: client,
	}
	return nil
}
